/*
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

/*
 *
 *
 *
 *
 *
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

package java.util.concurrent;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.*;

/**
 * A {@link ThreadPoolExecutor} that can additionally schedule
 * commands to run after a given delay, or to execute
 * periodically. This class is preferable to {@link java.util.Timer}
 * when multiple worker threads are needed, or when the additional
 * flexibility or capabilities of {@link ThreadPoolExecutor} (which
 * this class extends) are required.
 *
 * <p>Delayed tasks execute no sooner than they are enabled, but
 * without any real-time guarantees about when, after they are
 * enabled, they will commence. Tasks scheduled for exactly the same
 * execution time are enabled in first-in-first-out (FIFO) order of
 * submission.
 *
 * <p>When a submitted task is cancelled before it is run, execution
 * is suppressed. By default, such a cancelled task is not
 * automatically removed from the work queue until its delay
 * elapses. While this enables further inspection and monitoring, it
 * may also cause unbounded retention of cancelled tasks. To avoid
 * this, set {@link #setRemoveOnCancelPolicy} to {@code true}, which
 * causes tasks to be immediately removed from the work queue at
 * time of cancellation.
 *
 * <p>Successive executions of a task scheduled via
 * {@code scheduleAtFixedRate} or
 * {@code scheduleWithFixedDelay} do not overlap. While different
 * executions may be performed by different threads, the effects of
 * prior executions <a
 * href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
 * those of subsequent ones.
 *
 * <p>While this class inherits from {@link ThreadPoolExecutor}, a few
 * of the inherited tuning methods are not useful for it. In
 * particular, because it acts as a fixed-sized pool using
 * {@code corePoolSize} threads and an unbounded queue, adjustments
 * to {@code maximumPoolSize} have no useful effect. Additionally, it
 * is almost never a good idea to set {@code corePoolSize} to zero or
 * use {@code allowCoreThreadTimeOut} because this may leave the pool
 * without threads to handle tasks once they become eligible to run.
 *
 * <p><b>Extension notes:</b> This class overrides the
 * {@link ThreadPoolExecutor#execute(Runnable) execute} and
 * {@link AbstractExecutorService#submit(Runnable) submit}
 * methods to generate internal {@link ScheduledFuture} objects to
 * control per-task delays and scheduling.  To preserve
 * functionality, any further overrides of these methods in
 * subclasses must invoke superclass versions, which effectively
 * disables additional task customization.  However, this class
 * provides alternative protected extension method
 * {@code decorateTask} (one version each for {@code Runnable} and
 * {@code Callable}) that can be used to customize the concrete task
 * types used to execute commands entered via {@code execute},
 * {@code submit}, {@code schedule}, {@code scheduleAtFixedRate},
 * and {@code scheduleWithFixedDelay}.  By default, a
 * {@code ScheduledThreadPoolExecutor} uses a task type extending
 * {@link FutureTask}. However, this may be modified or replaced using
 * subclasses of the form:
 *
 *  <pre> {@code
 * public class CustomScheduledExecutor extends ScheduledThreadPoolExecutor {
 *
 *   static class CustomTask<V> implements RunnableScheduledFuture<V> { ... }
 *
 *   protected <V> RunnableScheduledFuture<V> decorateTask(
 *                Runnable r, RunnableScheduledFuture<V> task) {
 *       return new CustomTask<V>(r, task);
 *   }
 *
 *   protected <V> RunnableScheduledFuture<V> decorateTask(
 *                Callable<V> c, RunnableScheduledFuture<V> task) {
 *       return new CustomTask<V>(c, task);
 *   }
 *   // ... add constructors, etc.
 * }}</pre>
 *
 * @since 1.5
 * @author Doug Lea
 */
public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    /*
     * This class specializes ThreadPoolExecutor implementation by
     *
     * 1. Using a custom task type, ScheduledFutureTask for
     *    tasks, even those that don't require scheduling (i.e.,
     *    those submitted using ExecutorService execute, not
     *    ScheduledExecutorService methods) which are treated as
     *    delayed tasks with a delay of zero.
     *
     * 2. Using a custom queue (DelayedWorkQueue), a variant of
     *    unbounded DelayQueue. The lack of capacity constraint and
     *    the fact that corePoolSize and maximumPoolSize are
     *    effectively identical simplifies some execution mechanics
     *    (see delayedExecute) compared to ThreadPoolExecutor.
     *
     * 3. Supporting optional run-after-shutdown parameters, which
     *    leads to overrides of shutdown methods to remove and cancel
     *    tasks that should NOT be run after shutdown, as well as
     *    different recheck logic when task (re)submission overlaps
     *    with a shutdown.
     *
     * 4. Task decoration methods to allow interception and
     *    instrumentation, which are needed because subclasses cannot
     *    otherwise override submit methods to get this effect. These
     *    don't have any impact on pool control logic though.
     */

    /**
     * 线程池状态为 shutDown的时候  线程从任务队列内获取到  周期任务时  是否执行
     * false表示不执行
     * true表示  线程读取到周期任务时  还会执行一次
     *
     * False if should cancel/suppress periodic tasks on shutdown.
     */
    private volatile boolean continueExistingPeriodicTasksAfterShutdown;

    /**
     * 线程池状态为 shutDown的时候  线程从任务队列内获取到  延迟任务时  是否执行
     * false表示不执行
     * true表示执行
     *
     * False if should cancel non-periodic tasks on shutdown.
     */
    private volatile boolean executeExistingDelayedTasksAfterShutdown = true;

    /**
     * 任务被取消时  是否需要从任务队列移除
     * 默认不移除，等到线程拿到任务之后丢弃
     *
     * True if ScheduledFutureTask.cancel should remove from queue
     */
    private volatile boolean removeOnCancel = false;

    /**
     * 生成任务序列号的一个计数器
     * 为什么需要序列号呢？ 因为调度线程池内部使用的其实是 优先级队列  优先级队列需要实现 compare接口
     * 如果两个任务 通过compare之后 =0 那么就表示两个任务优先级一致，这个时候就需要使用  计数器来对比他们两个那个优先级更高
     *
     * Sequence number to break scheduling ties, and in turn to
     * guarantee FIFO order among tied entries.
     */
    private static final AtomicLong sequencer = new AtomicLong();

    /**
     * Returns current nanosecond time.
     */
    final long now() {
        return System.nanoTime();
    }

    /**
     * 调度线程池内部核心
     */
    private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {

        /**
         * 序列号  用于比较任务的优先级  当任务的time字段比较不出来的时候
         * 使用该字段进行比较  该字段一定不会相同
         * Sequence number to break ties FIFO
         */
        private final long sequenceNumber;

        /**
         * 任务下一次执行的时间节点  交付执行
         * 当任务执行结束之后  会再次提交任务到队列中  加入到queue之前会修改该字段为下一次执行时间点
         * The time the task is enabled to execute in nanoTime units
         */
        private long time;

        /**
         * 周期时间长度
         * >0的时候  表示当前任务是由  scheduleAtFixedRate 方法提交的   该任务不考虑执行耗时
         * <0的时候  表示当前任务是由  scheduleWithFixedDelay 方法提交的   该任务考虑执行耗时,下次执行节点为 本次结束时间+period
         * =0的时候  表示当前任务是由  schedule 方法提交的  该任务是一个延迟执行的任务   只执行一次
         *
         * Period in nanoseconds for repeating tasks.  A positive
         * value indicates fixed-rate execution.  A negative value
         * indicates fixed-delay execution.  A value of 0 indicates a
         * non-repeating task.
         */
        private final long period;

        /**
         * 指向自身
         * The actual task to be re-enqueued by reExecutePeriodic
         */
        RunnableScheduledFuture<V> outerTask = this;

        /**
         * 因为其内部的阻塞队列使用的是  堆（最小堆）  记录当前任务在堆中的索引
         * 堆是一个满二叉树，满二叉树一般基于数组实现  所以这个index相当于数组中的索引
         * Index into delay queue, to support faster cancellation.
         */
        int heapIndex;

        /**
         * Creates a one-shot action with given nanoTime-based trigger time.
         */
        ScheduledFutureTask(Runnable r, V result, long ns) {
            super(r, result);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a periodic action with given nano time and period.
         */
        ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a one-shot action with given nanoTime-based trigger time.
         */
        ScheduledFutureTask(Callable<V> callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * 返回当前任务还差多久下一次执行
         */
        public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), NANOSECONDS);
        }


        /**
         * 比较两个延迟任务的执行优先级
         * 同为ScheduledFutureTask的情况下
         * 优先比较 time，time大的执行时间比较久所以 比较小
         * 然后比较  序列号sequenceNumber
         *
         * 如果不是ScheduledFutureTask 则通过顶层接口的方法getDelay
         * 比较对应的延迟时间然后判断
         */
        public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                return 0;
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
                else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }

        /**
         * period = 0 表示当前任务不是周期（延迟任务）
         * 当 period！= 0时  该任务一定是一个周期任务
         *
         * Returns {@code true} if this is a periodic (not a one-shot) action.
         *
         * @return {@code true} if periodic
         */
        public boolean isPeriodic() {
            return period != 0;
        }

        /**
         * 周期任务执行之后，再次提交到 延迟队列之前调用的
         * 设置下一次任务执行的时间
         *
         * Sets the next time to run for a periodic task.
         */
        private void setNextRunTime() {
            long p = period;
            if (p > 0)
                //条件成立  表示当前任务是由 scheduleAtFixedRate 提交的,该任务不考虑执行耗时，所以直接用上一次时间+执行周期
                time += p;
            else
                //条件成立  表示当前任务是由 scheduleWithFixedDelay 提交的,该任务考虑执行耗时
                //使用系统当前时间 + 执行周期作为下一次执行的  时间节点
                time = triggerTime(-p);
        }

        /**
         * 取消任务
         * mayInterruptIfRunning  当前任务处于执行中时，是否需要终端
         * true  会先去中断任务
         * false  则直接设置任务状态为取消状态
         */
        public boolean cancel(boolean mayInterruptIfRunning) {
            //调用futureTask的 取消方法  cancelled表示任务是否取消成功
            //取消失败的情况:
            //1、任务已经执行完成
            //2、任务出现异常
            //3、任务已经被取消
            boolean cancelled = super.cancel(mayInterruptIfRunning);

            //条件成立  当前任务已经被取消   任务被取消时需要被移除    表示任务依然在堆中
            //heapIndex >= 0  条件成立  表示任务依然在堆中,如果从堆结构中移除的任务 heapIndex 会先设置为-1
            if (cancelled && removeOnCancel && heapIndex >= 0)
                //调用线程池的remove方法  最终触发的是  延迟队列的  remove方法
                remove(this);

            //返回取消结果
            return cancelled;
        }

        /**
         * 线程池中的方法调用 runnable的 run方法
         * runnable是 FutureTask的run方法
         * 这里重写自 FutureTask的run方法
         *
         * Overrides FutureTask version so as to reset/requeue if periodic.
         */
        public void run() {
            //判断当前任务是不是周期任务
            boolean periodic = isPeriodic();

            //条件成立： 当前任务不可继续执行
            if (!canRunInCurrentRunState(periodic))
                //取消当前任务
                cancel(false);
            //条件成立：  当前任务不是周期任务,只是简单的延迟任务
            else if (!periodic)
                //直接调用 FutureTask的 run方法 该方法执行完成之后会修改 FutureTask的state
                ScheduledFutureTask.super.run();
            //前置条件：  表示当前任务一定是一个周期任务
            //调用FutureTask的 runAndReset方法 该方法执行完成之后不会修改 FutureTask的state
            //返回 runAndReset方法 有且只有执行完成并且任务状态依然是 new的时候才会返回true
            else if (ScheduledFutureTask.super.runAndReset()) {
                //执行到这里  表示当前任务已经执行过了  重新设置当前任务的下一次的执行时间
                setNextRunTime();
                //把当前任务 加入到延迟队列中
                reExecutePeriodic(outerTask);
            }
        }
    }

    /**
     * 返回当前任务是否可以继续执行
     *
     * Returns true if can run a task given current run state
     * and run-after-shutdown parameters.
     *
     * @param periodic true if this task periodic, false if delayed
     */
    boolean canRunInCurrentRunState(boolean periodic) {
        //isRunningOrShutdown
        //线程池如果是running状态返回true
        //如果线程池是 shutdown状态 返回入参
        //也就是如果你允许  shutdown执行那么会返回true  如果不是则返回false

        //periodic ? continueExistingPeriodicTasksAfterShutdown : executeExistingDelayedTasksAfterShutdown
        //这个就是 两种不同任务在 shutdown状态下是否继续执行
        return isRunningOrShutdown(periodic ?
                                   continueExistingPeriodicTasksAfterShutdown :
                                   executeExistingDelayedTasksAfterShutdown);
    }

    /**
     * Main execution method for delayed or periodic tasks.  If pool
     * is shut down, rejects the task. Otherwise adds task to queue
     * and starts a thread, if necessary, to run it.  (We cannot
     * prestart the thread to run the task because the task (probably)
     * shouldn't be run yet.)  If the pool is shut down while the task
     * is being added, cancel and remove it if required by state and
     * run-after-shutdown parameters.
     *
     * @param task the task
     */
    private void delayedExecute(RunnableScheduledFuture<?> task) {
        if (isShutdown())
            reject(task);
        else {
            super.getQueue().add(task);
            if (isShutdown() &&
                !canRunInCurrentRunState(task.isPeriodic()) &&
                remove(task))
                task.cancel(false);
            else
                ensurePrestart();
        }
    }

    /**
     * 如果当前任务可以被执行
     * 那么会添加当前任务到队列中
     * 添加完成之后
     * 如果可以继续执行  那么会调用线程池的 ensurePrestart 方法 确保有线程可以执行任务
     * 如果任务无法继续执行且可以从线程池的阻塞队列中被移除 那么会调用 FutureTask的cancel方法来修改任务状态
     *
     * Requeues a periodic task unless current run state precludes it.
     * Same idea as delayedExecute except drops task rather than rejecting.
     *
     * @param task the task
     */
    void reExecutePeriodic(RunnableScheduledFuture<?> task) {
        //条件成立 表示当前任务可以被执行
        if (canRunInCurrentRunState(true)) {
            //将任务再次加入到任务队列中
            super.getQueue().add(task);
            //条件成立： 当前任务在线程池中已经无法继续执行  并且从线程池的阻塞队列中移除任务已经成功
            if (!canRunInCurrentRunState(true) && remove(task))
                //调用 FutureTask的  cancel  取消当前任务  把任务状态设置为取消状态
                task.cancel(false);
            //前置条件： 当前任务要么可以继续执行 要么依然在线程池的任务队列中无法被移除
            //也就是  要么正常状态  要么得继续执行完这一次
            else
                //线程池的一个方法  在特殊情况下用于补充线程数量
                //1、线程数少于核心线程数时
                //2、线程池未达到最大线程数且任务队列已经满时
                //确保任务提交之后 可以有线程执行
                ensurePrestart();
        }
    }

    /**
     * Cancels and clears the queue of all tasks that should not be run
     * due to shutdown policy.  Invoked within super.shutdown.
     */
    @Override void onShutdown() {
        BlockingQueue<Runnable> q = super.getQueue();
        boolean keepDelayed =
            getExecuteExistingDelayedTasksAfterShutdownPolicy();
        boolean keepPeriodic =
            getContinueExistingPeriodicTasksAfterShutdownPolicy();
        if (!keepDelayed && !keepPeriodic) {
            for (Object e : q.toArray())
                if (e instanceof RunnableScheduledFuture<?>)
                    ((RunnableScheduledFuture<?>) e).cancel(false);
            q.clear();
        }
        else {
            // Traverse snapshot to avoid iterator exceptions
            for (Object e : q.toArray()) {
                if (e instanceof RunnableScheduledFuture) {
                    RunnableScheduledFuture<?> t =
                        (RunnableScheduledFuture<?>)e;
                    if ((t.isPeriodic() ? !keepPeriodic : !keepDelayed) ||
                        t.isCancelled()) { // also remove if already cancelled
                        if (q.remove(t))
                            t.cancel(false);
                    }
                }
            }
        }
        tryTerminate();
    }

    /**
     * Modifies or replaces the task used to execute a runnable.
     * This method can be used to override the concrete
     * class used for managing internal tasks.
     * The default implementation simply returns the given task.
     *
     * @param runnable the submitted Runnable
     * @param task the task created to execute the runnable
     * @param <V> the type of the task's result
     * @return a task that can execute the runnable
     * @since 1.6
     */
    protected <V> RunnableScheduledFuture<V> decorateTask(
        Runnable runnable, RunnableScheduledFuture<V> task) {
        return task;
    }

    /**
     * Modifies or replaces the task used to execute a callable.
     * This method can be used to override the concrete
     * class used for managing internal tasks.
     * The default implementation simply returns the given task.
     *
     * @param callable the submitted Callable
     * @param task the task created to execute the callable
     * @param <V> the type of the task's result
     * @return a task that can execute the callable
     * @since 1.6
     */
    protected <V> RunnableScheduledFuture<V> decorateTask(
        Callable<V> callable, RunnableScheduledFuture<V> task) {
        return task;
    }

    /**
     * Creates a new {@code ScheduledThreadPoolExecutor} with the
     * given core pool size.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

    /**
     * Creates a new {@code ScheduledThreadPoolExecutor} with the
     * given initial parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     * @throws NullPointerException if {@code threadFactory} is null
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     * initial parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     * @throws NullPointerException if {@code handler} is null
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), handler);
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     * initial parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     * @throws NullPointerException if {@code threadFactory} or
     *         {@code handler} is null
     *
     * corePoolSize  核心线程数
     * threadFactory   线程工厂
     * handler      拒绝策略
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory,
                                       RejectedExecutionHandler handler) {
        //最大线程数   Integer.MAX_VALUE
        //非核心线程  一空闲即销毁   0, NANOSECONDS（线程其实是部分核心与非核心的，只是线程会维护一个 corePoolSize的线程数量）
        //DelayedWorkQueue  内部自己实现的一个  任务队列，延迟优先级队列，在优先级的基础上提供了延迟的功能，提供了hold线程的功能
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory, handler);
    }

    /**
     * Returns the trigger time of a delayed action.
     */
    private long triggerTime(long delay, TimeUnit unit) {
        return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
    }

    /**
     * 计算下一次执行的时间
     *
     * Returns the trigger time of a delayed action.
     */
    long triggerTime(long delay) {
        return now() +
                //如果 delay时间超过Long的一半  那么需要走一个防止时间溢出的处理
            ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
    }

    /**
     * 为了防止延迟时间溢出，所以进行特殊处理
     * todo 待完善
     *
     * Constrains the values of all delays in the queue to be within
     * Long.MAX_VALUE of each other, to avoid overflow in compareTo.
     * This may occur if a task is eligible to be dequeued, but has
     * not yet been, while some other task is added with a delay of
     * Long.MAX_VALUE.
     */
    private long overflowFree(long delay) {
        //获取当前延迟队列的  对头
        Delayed head = (Delayed) super.getQueue().peek();
        if (head != null) {
            long headDelay = head.getDelay(NANOSECONDS);
            if (headDelay < 0 && (delay - headDelay < 0))
                delay = Long.MAX_VALUE + headDelay;
        }
        //队头为空  可以理解为当前状态只会有一个任务  所以不需要做特殊处理
        return delay;
    }

    /**
     * 调度线程池的提交方法  提交的任务为  普通延迟任务
     * 延迟时间为   delay + unit决定  并且只执行一次
     *
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Void>(command, null,
                                          triggerTime(delay, unit)));
        delayedExecute(t);
        return t;
    }

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        if (callable == null || unit == null)
            throw new NullPointerException();
        RunnableScheduledFuture<V> t = decorateTask(callable,
            new ScheduledFutureTask<V>(callable,
                                       triggerTime(delay, unit)));
        delayedExecute(t);
        return t;
    }

    /**
     * 提交的任务为 周期任务  按照一个周期反复执行
     * command会在  initialDelay的unit之后   按照period的unit间隔去执行
     *
     * 该方法提交的任务  不考虑执行的耗时 就比如：
     * 03：00：00开始执行第一次  执行间隔10秒
     * 如果执行时间小于10秒  那么下一次任务会在 03：00：10时执行
     * 如果执行时间大于10秒  那么下一次任务就是立马执行
     *
     *
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     * @throws IllegalArgumentException   {@inheritDoc}
     */
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(period));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

    /**
     * 提交的任务为 周期任务  按照一个周期反复执行
     * command会在  initialDelay的unit之后   按照period的unit间隔去执行
     *
     * 该方法提交的任务  会考虑执行的耗时 就比如：
     * 03：00：00开始执行第一次  执行间隔10秒
     * 如果执行时间小于10秒，以2秒为例  那么下一次任务会在 03：00：12时执行
     * 如果执行时间大于10秒，以20秒为例  那么下一次任务会在 03：00：30时执行
     *
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     * @throws IllegalArgumentException   {@inheritDoc}
     */
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (delay <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

    /**
     * Executes {@code command} with zero required delay.
     * This has effect equivalent to
     * {@link #schedule(Runnable,long,TimeUnit) schedule(command, 0, anyUnit)}.
     * Note that inspections of the queue and of the list returned by
     * {@code shutdownNow} will access the zero-delayed
     * {@link ScheduledFuture}, not the {@code command} itself.
     *
     * <p>A consequence of the use of {@code ScheduledFuture} objects is
     * that {@link ThreadPoolExecutor#afterExecute afterExecute} is always
     * called with a null second {@code Throwable} argument, even if the
     * {@code command} terminated abruptly.  Instead, the {@code Throwable}
     * thrown by such a task can be obtained via {@link Future#get}.
     *
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution because the
     *         executor has been shut down
     * @throws NullPointerException {@inheritDoc}
     */
    public void execute(Runnable command) {
        schedule(command, 0, NANOSECONDS);
    }

    // Override AbstractExecutorService methods

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public Future<?> submit(Runnable task) {
        return schedule(task, 0, NANOSECONDS);
    }

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Runnable task, T result) {
        return schedule(Executors.callable(task, result), 0, NANOSECONDS);
    }

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Callable<T> task) {
        return schedule(task, 0, NANOSECONDS);
    }

    /**
     * Sets the policy on whether to continue executing existing
     * periodic tasks even when this executor has been {@code shutdown}.
     * In this case, these tasks will only terminate upon
     * {@code shutdownNow} or after setting the policy to
     * {@code false} when already shutdown.
     * This value is by default {@code false}.
     *
     * @param value if {@code true}, continue after shutdown, else don't
     * @see #getContinueExistingPeriodicTasksAfterShutdownPolicy
     */
    public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
        continueExistingPeriodicTasksAfterShutdown = value;
        if (!value && isShutdown())
            onShutdown();
    }

    /**
     * Gets the policy on whether to continue executing existing
     * periodic tasks even when this executor has been {@code shutdown}.
     * In this case, these tasks will only terminate upon
     * {@code shutdownNow} or after setting the policy to
     * {@code false} when already shutdown.
     * This value is by default {@code false}.
     *
     * @return {@code true} if will continue after shutdown
     * @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
     */
    public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
        return continueExistingPeriodicTasksAfterShutdown;
    }

    /**
     * Sets the policy on whether to execute existing delayed
     * tasks even when this executor has been {@code shutdown}.
     * In this case, these tasks will only terminate upon
     * {@code shutdownNow}, or after setting the policy to
     * {@code false} when already shutdown.
     * This value is by default {@code true}.
     *
     * @param value if {@code true}, execute after shutdown, else don't
     * @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
        executeExistingDelayedTasksAfterShutdown = value;
        if (!value && isShutdown())
            onShutdown();
    }

    /**
     * Gets the policy on whether to execute existing delayed
     * tasks even when this executor has been {@code shutdown}.
     * In this case, these tasks will only terminate upon
     * {@code shutdownNow}, or after setting the policy to
     * {@code false} when already shutdown.
     * This value is by default {@code true}.
     *
     * @return {@code true} if will execute after shutdown
     * @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
        return executeExistingDelayedTasksAfterShutdown;
    }

    /**
     * Sets the policy on whether cancelled tasks should be immediately
     * removed from the work queue at time of cancellation.  This value is
     * by default {@code false}.
     *
     * @param value if {@code true}, remove on cancellation, else don't
     * @see #getRemoveOnCancelPolicy
     * @since 1.7
     */
    public void setRemoveOnCancelPolicy(boolean value) {
        removeOnCancel = value;
    }

    /**
     * Gets the policy on whether cancelled tasks should be immediately
     * removed from the work queue at time of cancellation.  This value is
     * by default {@code false}.
     *
     * @return {@code true} if cancelled tasks are immediately removed
     *         from the queue
     * @see #setRemoveOnCancelPolicy
     * @since 1.7
     */
    public boolean getRemoveOnCancelPolicy() {
        return removeOnCancel;
    }

    /**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted.
     * Invocation has no additional effect if already shut down.
     *
     * <p>This method does not wait for previously submitted tasks to
     * complete execution.  Use {@link #awaitTermination awaitTermination}
     * to do that.
     *
     * <p>If the {@code ExecuteExistingDelayedTasksAfterShutdownPolicy}
     * has been set {@code false}, existing delayed tasks whose delays
     * have not yet elapsed are cancelled.  And unless the {@code
     * ContinueExistingPeriodicTasksAfterShutdownPolicy} has been set
     * {@code true}, future executions of existing periodic tasks will
     * be cancelled.
     *
     * @throws SecurityException {@inheritDoc}
     */
    public void shutdown() {
        super.shutdown();
    }

    /**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks
     * that were awaiting execution.
     *
     * <p>This method does not wait for actively executing tasks to
     * terminate.  Use {@link #awaitTermination awaitTermination} to
     * do that.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  This implementation
     * cancels tasks via {@link Thread#interrupt}, so any task that
     * fails to respond to interrupts may never terminate.
     *
     * @return list of tasks that never commenced execution.
     *         Each element of this list is a {@link ScheduledFuture},
     *         including those tasks submitted using {@code execute},
     *         which are for scheduling purposes used as the basis of a
     *         zero-delay {@code ScheduledFuture}.
     * @throws SecurityException {@inheritDoc}
     */
    public List<Runnable> shutdownNow() {
        return super.shutdownNow();
    }

    /**
     * Returns the task queue used by this executor.  Each element of
     * this queue is a {@link ScheduledFuture}, including those
     * tasks submitted using {@code execute} which are for scheduling
     * purposes used as the basis of a zero-delay
     * {@code ScheduledFuture}.  Iteration over this queue is
     * <em>not</em> guaranteed to traverse tasks in the order in
     * which they will execute.
     *
     * @return the task queue
     */
    public BlockingQueue<Runnable> getQueue() {
        return super.getQueue();
    }

    /**
     * 优先级 延迟 阻塞队列    其实现方式为堆
     * 如果当前任务没有到时间点可以执行的话  会被当前队列hold挂起之类的处理
     *
     * 堆分为  最大堆(大顶堆、大根堆)   最小堆(小顶堆、小根堆)  其数据结构的实现  是以满二叉树来表示的
     * 以最小堆为例: 数组为:[1,2,3,4,5,6,7]
     * 公式有：
     * 1、查询父节点：  floor( (i-1) / 2 )
     * 2、查询左子节点：   left = i * 2 + 1
     * 3、查询右子节点：   right = i * 2 + 2
     *
     * 新元素的插入： https://www.cs.usfca.edu/~galles/visualization/Heap.html
     * 1、先插入到数组的最后一个位置
     * 2、和父节点进行比较，  如果优先级高于父节点那么两个节点交换
     * 3、重复步骤2  直到  新元素无法继续向上交换为止
     *
     * 顶层元素的删除流程：
     * 1、把数组[0]号位置的元素删除
     * 2、把数组最后一个位置的元素移动到[0]位置来  并且把最后一个位置设置为null
     * 3、选择两个子节点中优先级最高的那个进行比较  如果没有其中一个优先级高那么就交换
     * 4、重复步骤3  直到当前元素无法继续下沉为止
     *
     * eg:   最小堆
     *       1
     *      /  \
     *     2    3
     *    / \  / \
     *   4  5 6  7
     * eg：   最大堆
     *       10
     *      /  \
     *     9    8
     *    / \  / \
     *   4  5 6  7
     *
     * 当前队列采用的堆是  最小堆  最底层的元素  优先级最高
     *
     *
     * Specialized delay queue. To mesh with TPE declarations, this
     * class must be declared as a BlockingQueue<Runnable> even though
     * it can only hold RunnableScheduledFutures.
     */
    static class DelayedWorkQueue extends AbstractQueue<Runnable>
        implements BlockingQueue<Runnable> {

        /*
         * A DelayedWorkQueue is based on a heap-based data structure
         * like those in DelayQueue and PriorityQueue, except that
         * every ScheduledFutureTask also records its index into the
         * heap array. This eliminates the need to find a task upon
         * cancellation, greatly speeding up removal (down from O(n)
         * to O(log n)), and reducing garbage retention that would
         * otherwise occur by waiting for the element to rise to top
         * before clearing. But because the queue may also hold
         * RunnableScheduledFutures that are not ScheduledFutureTasks,
         * we are not guaranteed to have such indices available, in
         * which case we fall back to linear search. (We expect that
         * most tasks will not be decorated, and that the faster cases
         * will be much more common.)
         *
         * All heap operations must record index changes -- mainly
         * within siftUp and siftDown. Upon removal, a task's
         * heapIndex is set to -1. Note that ScheduledFutureTasks can
         * appear at most once in the queue (this need not be true for
         * other kinds of tasks or work queues), so are uniquely
         * identified by heapIndex.
         */

        /**
         * 初始化时  数组的大小
         */
        private static final int INITIAL_CAPACITY = 16;

        /**
         * 这个数组就是满二叉树  用于存储提交的任务
         */
        private RunnableScheduledFuture<?>[] queue =
            new RunnableScheduledFuture<?>[INITIAL_CAPACITY];

        /**
         * 全局的可重入锁
         * 添加和删除操作的时候都需要加锁操作  以确保线程安全
         */
        private final ReentrantLock lock = new ReentrantLock();

        /**
         * 当前队列中存储的任务数量
         */
        private int size = 0;

        /**
         * Thread designated to wait for the task at the head of the
         * queue.  This variant of the Leader-Follower pattern
         * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
         * minimize unnecessary timed waiting.  When a thread becomes
         * the leader, it waits only for the next delay to elapse, but
         * other threads await indefinitely.  The leader thread must
         * signal some other thread before returning from take() or
         * poll(...), unless some other thread becomes leader in the
         * interim.  Whenever the head of the queue is replaced with a
         * task with an earlier expiration time, the leader field is
         * invalidated by being reset to null, and some waiting
         * thread, but not necessarily the current leader, is
         * signalled.  So waiting threads must be prepared to acquire
         * and lose leadership while waiting.
         *
         * 线程池内的某个线程使用 take方法  获取任务的时候  如果堆顶元素不为空/null 但是堆顶的任务还不到可以执行的时间
         * 这个时候  线程就回去检查队列的  leader字段是否被占用
         *
         * 未被占用：
         * 将当前线程保存到 leader字段中，然后再把当前线程 以  堆顶任务.time-now()的时间为超时时间  挂起到 available条件队列中
         *
         * 被占用：
         * 直接以不指定超时时间的方式挂起到  available条件队列中
         *
         * 那么不指定超时时间什么时候会被唤醒呢？
         * 既然leader字段已经有值了，那么表示 available条件队列中已经有线程以  指定了超时时间的方式挂起了  所以leader字段的线程会优先醒过来
         * 醒来之后 leader字段的线程就会从 堆顶拿走任务去执行  拿走任务之后  会释放lock锁  释放lock锁之前会做一件事也就是  available.signal
         * 去唤醒后面的线程，后面的线程被唤醒之后  会去哪儿？  会回到AQS的队列中  然后尝试获取  lock锁
         *
         */
        private Thread leader = null;

        /**
         * Condition signalled when a newer task becomes available at the
         * head of the queue or a new thread may need to become leader.
         *
         * 条件队列
         */
        private final Condition available = lock.newCondition();

        /**
         * Sets f's heapIndex if it is a ScheduledFutureTask.
         *
         * 设置 调度任务的 heapIndex 字段
         */
        private void setIndex(RunnableScheduledFuture<?> f, int idx) {
            if (f instanceof ScheduledFutureTask)
                ((ScheduledFutureTask)f).heapIndex = idx;
        }

        /**
         * Sifts element added at bottom up to its heap-ordered spot.
         * Call only when holding lock.
         *
         * 向上冒泡
         * k  新插入元素的下标
         * key   新提交的任务
         *
         */
        private void siftUp(int k, RunnableScheduledFuture<?> key) {
            //k ==0 表示什么情况呢？  就是当前元素已经上移到 最前面了  没法继续移动了
            while (k > 0) {
                //获取到当前  元素的父节点下标   (k-1)/2
                int parent = (k - 1) >>> 1;
                //获取当前元素的 父节点
                RunnableScheduledFuture<?> e = queue[parent];
                //条件成立  当前元素优先级没有父节点高  那么直接break不需要 更新下标
                if (key.compareTo(e) >= 0)
                    break;

                //执行到这里  表示  新提交的任务  优先级高于其父节点  所以需要进行替换
                //把父节点下移
                queue[k] = e;
                //修改父节点的 下标
                setIndex(e, k);
                //更新下标
                k = parent;
            }

            //执行到这里 表示当前任务  已经找到了目标slot可以进行插入操作了
            //当前元素入队到指定位置
            queue[k] = key;
            //修改当前节点的下标
            setIndex(key, k);
        }

        /**
         * Sifts element added at top down to its heap-ordered spot.
         * Call only when holding lock.
         *
         * 向下冒泡
         * k   当前开始向下坠落的位置
         * key    堆最后的节点
         */
        private void siftDown(int k, RunnableScheduledFuture<?> key) {
            //half 是做什么的呢？  是作为  while循环的一个判断条件的
            //来到这里之前  size之前已经-1过了
            //而k这个值  最重要变成  key对象存储的  slot下标的
            int half = size >>> 1;

            //为什么会有这个判断呢？
            //在堆这个满二叉树的结构中  执行到这里表示已经有元素出队了，这个时候  size/2是什么位置呢？
            //其实是 当前满二叉树最底层的最左一个节点的下标  如果  下坠的位置 >=这个坐标那么会发生什么呢？
            //那其实就是  当前节点不需要往下坠了，这个节点已经被坠到最后一层了  没有必要往下走了
            //所以这个  while条件就这么判断了
            while (k < half) {
                //获取当前k元素的左子节点下标
                //而child最终表示  key的左右子节点中  优先级高的 子节点 下标
                int child = (k << 1) + 1;

                //获取左子节点元素
                RunnableScheduledFuture<?> c = queue[child];
                //获取右子节点下标
                int right = child + 1;
                //条件成立：   右子节点是有效的并且  优先级高于左子节点
                if (right < size && c.compareTo(queue[right]) > 0)
                    //更新c的值   顺便更新child的值
                    c = queue[child = right];

                //条件成立：  当前key的优先级是高于  子节点的  这个时候表示已经下坠的最终位置了不需要继续下坠了
                if (key.compareTo(c) <= 0)
                    break;

                //执行到这里  表示  key的优先级低于  子节点 需要继续下坠
                //需要把优先级高的子节点与当前节点交换位置

                //让c占用当前 key的位置
                queue[k] = c;
                //重新设置c的 slot下标
                setIndex(c, k);
                //更新k的值
                k = child;
            }
            //执行到这里表示 已经下坠到目标位置可以进行  赋值了

            //目标slot位置赋值
            queue[k] = key;
            //设置key的下标
            setIndex(key, k);
        }

        /**
         * Resizes the heap array.  Call only when holding lock.
         * 数组扩容的方法  每次扩容1.5
         */
        private void grow() {
            //获取当前数组长度
            int oldCapacity = queue.length;
            //1 + 0.5  扩容为原来的1.5倍
            int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
            //极限值判断
            if (newCapacity < 0) // overflow
                newCapacity = Integer.MAX_VALUE;
            //数组值 复制
            queue = Arrays.copyOf(queue, newCapacity);
        }

        /**
         * Finds index of given object, or -1 if absent.
         * 如果x 在任务队列内  则返回  其数组下标
         * 如果不在则返回-1
         */
        private int indexOf(Object x) {
            if (x != null) {
                if (x instanceof ScheduledFutureTask) {
                    //当传入的是  ScheduledFutureTask 任务时  使用其数组下标作为最终返回结果
                    //但返回之前会校验该下标是否生效
                    int i = ((ScheduledFutureTask) x).heapIndex;
                    // Sanity check; x could conceivably be a
                    // ScheduledFutureTask from some other pool.
                    //下标在数组内  不大于数组最大长度  不小于0  且目标位置的元素与传入元素相同
                    //则认为当前下标处于生效状态
                    if (i >= 0 && i < size && queue[i] == x)
                        return i;
                } else {
                    //当传入的 不是ScheduledFutureTask 任务时 遍历数组 遇到相同的则返回
                    for (int i = 0; i < size; i++)
                        if (x.equals(queue[i]))
                            return i;
                }
            }
            return -1;
        }

        /**
         * 检查指定x是否在队列内
         */
        public boolean contains(Object x) {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return indexOf(x) != -1;
            } finally {
                lock.unlock();
            }
        }

        /**
         *
         */
        public boolean remove(Object x) {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                int i = indexOf(x);
                if (i < 0)
                    return false;

                setIndex(queue[i], -1);
                int s = --size;
                RunnableScheduledFuture<?> replacement = queue[s];
                queue[s] = null;
                if (s != i) {
                    siftDown(i, replacement);
                    if (queue[i] == replacement)
                        siftUp(i, replacement);
                }
                return true;
            } finally {
                lock.unlock();
            }
        }

        public int size() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return size;
            } finally {
                lock.unlock();
            }
        }

        public boolean isEmpty() {
            return size() == 0;
        }

        public int remainingCapacity() {
            return Integer.MAX_VALUE;
        }

        /**
         * peek和 take的区别
         *
         * peek仅返回堆顶元素
         * 而take不仅会返回对顶元素 还会把对顶元素从堆中移除
         *
         */
        public RunnableScheduledFuture<?> peek() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return queue[0];
            } finally {
                lock.unlock();
            }
        }

        /**
         * 添加任务的核心方法
         *
         * x 一般就是 ScheduledFutureTask 的实例
         */
        public boolean offer(Runnable x) {
            if (x == null)
                throw new NullPointerException();
            //类型转换
            RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
            final ReentrantLock lock = this.lock;
            //加锁
            lock.lock();
            try {
                //获取当前  元素数量
                int i = size;
                //临界值判断  如果元素数量已经满了  甚至已经超过了 那么触发一次扩容
                if (i >= queue.length)
                    grow();
                //size++  记录当前元素数量
                size = i + 1;
                //条件成立  表示当前元素是第一个进入队列的元素
                if (i == 0) {
                    //第一个元素  直接翻进去  设置一下 index即可
                    queue[0] = e;
                    setIndex(e, 0);
                } else {
                    //非第一个元素都需要  向上冒泡  也就是移动到合适的位置
                    //i  表示 新插入元素的下标
                    //e  表示 新提交的任务
                    siftUp(i, e);
                }

                //条件成立的几种情况：
                //1、当前任务是第一个进入队列的任务
                //这个时候  leader字段本来就是空  设置为 null也没有什么影响
                //available.signal() 会唤醒一个等待任务的线程  让它尝试获取任务  如果没有到该执行的时间还是会被挂起
                //2、当前任务不是  第一个进入队列的任务  但是当前任务优先级特别高   它上升到了堆顶
                //这个情况下  leader字段可能是有值的  因为原来堆顶是有任务还没到该执行的时间   leader线程设置了超时时间在 available 挂起了
                //如果 leader字段有值得情况下  在这里阻塞等等待得线程就不应该继续等待了  毕竟他的等待时间和当前最优任务的时间不一定一致
                //相当于要重置一下   leader线程的挂起时间  所以清空leader线程  然后唤醒leader线程
                if (queue[0] == e) {
                    leader = null;
                    available.signal();
                }
            } finally {
                //锁释放
                lock.unlock();
            }
            //返回true  添加任务成功
            return true;
        }

        public void put(Runnable e) {
            offer(e);
        }

        public boolean add(Runnable e) {
            return offer(e);
        }

        public boolean offer(Runnable e, long timeout, TimeUnit unit) {
            return offer(e);
        }

        /**
         *
         * 完成当前任务出队  并平衡堆结构
         *
         * Performs common bookkeeping for poll and take: Replaces
         * first element with last and sifts it down.  Call only when
         * holding lock.
         * @param f the task to remove and return
         */
        private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
            //队列元素个数-1
            int s = --size;
            //获取最后一个元素
            RunnableScheduledFuture<?> x = queue[s];
            //
            queue[s] = null;
            //条件成立   最后一个元素不是堆顶元素
            if (s != 0)
                //这个情况下  默认最后一个元素为堆顶元素  然后走一个向下冒泡的逻辑
                //这个时候已经把 最后一个位置清空掉了 相当于让最后一个元素重新定一下自己的位置
                siftDown(0, x);
            //修改目标任务的下标
            setIndex(f, -1);
            //返回目标任务
            return f;
        }

        /**
         * 获取堆顶元素  有且只有堆顶元素到执行时间了才会返回  否则只会返回null
         * 注意：不会主动清空堆顶元素
         */
        public RunnableScheduledFuture<?> poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                //获取堆顶元素
                RunnableScheduledFuture<?> first = queue[0];
                //条件成立   对顶元素为空    或者    堆顶元素没有到执行时间
                if (first == null || first.getDelay(NANOSECONDS) > 0)
                    //这种情况下   直接返回null
                    return null;
                else
                    //执行到这里表示   当前堆顶元素已经到执行时间了   需要返回了   并且需要清空
                    return finishPoll(first);
            } finally {
                lock.unlock();
            }
        }

        /**
         * 核心获取元素的方法
         * 线程池在 核心数量以内时  <= corePoolSize的时候  线程获取任务使用的时  take()方法
         */
        public RunnableScheduledFuture<?> take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            //执行相应中断的  加锁方式
            lock.lockInterruptibly();
            try {
                //自旋
                for (;;) {
                    //获取当前堆顶元素
                    RunnableScheduledFuture<?> first = queue[0];

                    //堆顶元素为空直接进入阻塞等待
                    if (first == null)
                        //当第一个任务被offer进来之后  会唤醒第一个阻塞的节点
                        available.await();
                    else {
                        //执行到这里   堆顶元素不为空   可以尝试获取堆顶元素

                        //获取当前堆顶元素的  延迟执行时间
                        long delay = first.getDelay(NANOSECONDS);

                        //条件成立  当前任务的执行时间已经到点了   可以把任务移除堆并且返回了
                        if (delay <= 0)
                            return finishPoll(first);

                        //执行到这里  表示当前任务 有延迟时间还没到该执行的时间
                        //那怎么办呢？   首先得选择一个线程放到  leader字段中  然后再把没有放到leader字段中的线程挂起
                        first = null; // don't retain ref while waiting

                        //如果leader字段不为空  表示当前已经有线程   计时等待了   有线程可以唤醒自己
                        if (leader != null)
                            //所以  直接把当前线程挂起即可  为什么时不限时间的挂起呢?  这是因为没有任务提前醒来还要再次挂起
                            //那么什么时候 唤醒呢？   offer任务的时候   会在offer第一个任务的时候   唤醒挂起的第一个线程或者leader字段上的线程
                            //当第一个线程获取到任务之后会唤醒后续阻塞的线程的
                            available.await();
                        else {
                            //执行到这里  表示当前leader字段为空   当前线程是  第一个进入等待的线程

                            //获取当前线程
                            Thread thisThread = Thread.currentThread();
                            //赋值leader字段  这里不会有并发因为  外面已经加锁了
                            leader = thisThread;
                            try {
                                //当前线程定时挂起
                                //要么到时间唤醒
                                //要么有新的优先级更高的线程进来之后  唤醒了你
                                available.awaitNanos(delay);
                            } finally {
                                //条件成立  当前线程挂之后已经醒来了  但是发现leader还是自己  所以清空leader线程的引用
                                //因为 leader线程是为了保证等待那一群线程  可以被唤醒  这个时候  leader线程已经醒过来了
                                //可以  直接获取任务 然后唤醒后面的线程了
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                //条件成立：
                //leader == null   表示要么leader已经醒过来了    要么当前只有一个元素
                //queue[0] != null   表示当前堆中还有元素
                //以上两个条件成立   表示  当前leader已经从  阻塞中新过来了，这个时候需要唤醒被阻塞的其他线程了
                if (leader == null && queue[0] != null)
                    //唤醒被阻塞的其他线程
                    available.signal();
                //释放锁
                lock.unlock();
            }
        }

        /**
         * 核心拉去数据的方法
         * 线程池在获取任务 线程数量> corePoolSize时  使用该方法获取任务
         */
        public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit)
            throws InterruptedException {
            //获取获取任务的等待时间
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            //使用 相应中断的加锁方式枷锁  如果执行了中断就抛出异常
            lock.lockInterruptibly();
            try {
                //自旋
                for (;;) {
                    //获取堆顶元素
                    RunnableScheduledFuture<?> first = queue[0];

                    //条件成立  堆顶元素为空
                    if (first == null) {
                        //条件成立  如果等待时间已经消耗殆尽那么就 直接返回null
                        if (nanos <= 0)
                            return null;
                        else
                            //还有等待时间 执行有等待时间的挂起  并获取剩余等待时间
                            nanos = available.awaitNanos(nanos);
                    } else {
                        //执行到这里   表示堆顶元素不为空

                        //获取当前任务的延迟时间
                        long delay = first.getDelay(NANOSECONDS);

                        //延迟时间已经消耗殆尽  已经到了执行时间了
                        if (delay <= 0)
                            //获取堆顶元素并让堆顶元素出队
                            return finishPoll(first);

                        //执行到这里表示  任务没有到时间执行

                        //判断等待时间是否已经被消耗殆尽
                        if (nanos <= 0)
                            //已经没有等待时间  可以直接返回了  毕竟任务也没有到时间不需要继续往下走了
                            return null;

                        //
                        first = null; // don't retain ref while waiting

                        //条件1： nanos < delay   等待时间小于延迟时间
                        //条件2：  leader != null  当前堆顶元素没有线程在等待
                        //条件成立:  要么等待时间大于延迟时间  要么堆顶元素没有线程在等待
                        if (nanos < delay || leader != null)
                            //为什么  等待时间小于延迟时间也可以进来？
                            //因为堆顶元素可能是会变化的，这个时候让线程先去阻塞等待  如果堆中有优先级更高的任务进来
                            //那么他在 调用signal方法的时候  就可以唤醒当前阻塞在条件队列中的leader线程 leader线程可能会唤醒自己

                            //如果  等待时间大于延迟时间  为什么也要按照等待时间去阻塞？
                            //这是因为当前线程进去的时候  并不是第一个线程  但是自己又需要固定的时间阻塞
                            //那么就可以使用这个方法来处理 如果可以唤醒那么leader线程会把自己唤醒
                            nanos = available.awaitNanos(nanos);
                        else {
                            //执行到这里的前置条件    等待时间大于延迟时间  并且leader字段没有值
                            //换句话来说就是  当前线程是第一个等待的线程  这个时候自己就是leader  不存在等不到任务的情况  可以自由发挥了

                            //获取当前线程
                            Thread thisThread = Thread.currentThread();
                            //leader字段赋值
                            leader = thisThread;
                            try {
                                //尝试挂起延迟时间  并获剩余等待时间
                                long timeLeft = available.awaitNanos(delay);
                                // 延迟时间  减去  剩余等待时间  =   一共消耗了多少时间等待
                                // 等待时间  减去等待任务的时间  计算出剩余等待的时间
                                nanos -= delay - timeLeft;
                            } finally {
                                //如果线程唤醒的时候   自己还是leader线程那么就   清空leader线程字段
                                //为什么要这么做？
                                //因为阻塞可能会提前唤醒  也有可能会被中断唤醒
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                //条件成立：   leader字段为空  并且存在堆顶元素
                //简单来讲就是   当前没有线程在计时等待  并且还有任务可以获取
                if (leader == null && queue[0] != null)
                    //这个时候  尝试从条件队列中唤醒下一个线程来执行  尝试获取任务
                    available.signal();
                //释放重入锁
                lock.unlock();
            }
        }

        public void clear() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                for (int i = 0; i < size; i++) {
                    RunnableScheduledFuture<?> t = queue[i];
                    if (t != null) {
                        queue[i] = null;
                        setIndex(t, -1);
                    }
                }
                size = 0;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Returns first element only if it is expired.
         * Used only by drainTo.  Call only when holding lock.
         */
        private RunnableScheduledFuture<?> peekExpired() {
            // assert lock.isHeldByCurrentThread();
            RunnableScheduledFuture<?> first = queue[0];
            return (first == null || first.getDelay(NANOSECONDS) > 0) ?
                null : first;
        }

        public int drainTo(Collection<? super Runnable> c) {
            if (c == null)
                throw new NullPointerException();
            if (c == this)
                throw new IllegalArgumentException();
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                RunnableScheduledFuture<?> first;
                int n = 0;
                while ((first = peekExpired()) != null) {
                    c.add(first);   // In this order, in case add() throws.
                    finishPoll(first);
                    ++n;
                }
                return n;
            } finally {
                lock.unlock();
            }
        }

        public int drainTo(Collection<? super Runnable> c, int maxElements) {
            if (c == null)
                throw new NullPointerException();
            if (c == this)
                throw new IllegalArgumentException();
            if (maxElements <= 0)
                return 0;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                RunnableScheduledFuture<?> first;
                int n = 0;
                while (n < maxElements && (first = peekExpired()) != null) {
                    c.add(first);   // In this order, in case add() throws.
                    finishPoll(first);
                    ++n;
                }
                return n;
            } finally {
                lock.unlock();
            }
        }

        public Object[] toArray() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return Arrays.copyOf(queue, size, Object[].class);
            } finally {
                lock.unlock();
            }
        }

        @SuppressWarnings("unchecked")
        public <T> T[] toArray(T[] a) {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                if (a.length < size)
                    return (T[]) Arrays.copyOf(queue, size, a.getClass());
                System.arraycopy(queue, 0, a, 0, size);
                if (a.length > size)
                    a[size] = null;
                return a;
            } finally {
                lock.unlock();
            }
        }

        public Iterator<Runnable> iterator() {
            return new Itr(Arrays.copyOf(queue, size));
        }

        /**
         * Snapshot iterator that works off copy of underlying q array.
         */
        private class Itr implements Iterator<Runnable> {
            final RunnableScheduledFuture<?>[] array;
            int cursor = 0;     // index of next element to return
            int lastRet = -1;   // index of last element, or -1 if no such

            Itr(RunnableScheduledFuture<?>[] array) {
                this.array = array;
            }

            public boolean hasNext() {
                return cursor < array.length;
            }

            public Runnable next() {
                if (cursor >= array.length)
                    throw new NoSuchElementException();
                lastRet = cursor;
                return array[cursor++];
            }

            public void remove() {
                if (lastRet < 0)
                    throw new IllegalStateException();
                DelayedWorkQueue.this.remove(array[lastRet]);
                lastRet = -1;
            }
        }
    }
}
